import com.alibaba.fastjson.JSON
import org.apache.doris.flink.cfg.{DorisExecutionOptions, DorisOptions, DorisReadOptions}
import org.apache.doris.flink.deserialization.SimpleListDeserializationSchema
import org.apache.doris.flink.sink.DorisSink
import org.apache.doris.flink.sink.writer.{RowDataSerializer, SimpleStringSerializer}
import org.apache.doris.flink.source.DorisSourceBuilder
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.data.{GenericRowData, RowData, StringData}

import java.util.Properties


object FlinkDoris {

  def readAll(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val builder: DorisOptions.Builder = DorisOptions.builder()
      .setFenodes("192.168.206.130:8030")
      .setTableIdentifier("test_db.flinktest")
      .setUsername("root")
      .setPassword("")

    val dorisDource = DorisSourceBuilder.builder()
      .setDorisOptions(builder.build())
      .setDorisReadOptions(DorisReadOptions.builder().build())
      .setDeserializer(new SimpleListDeserializationSchema())
      .build()

    env.fromSource(dorisDource, WatermarkStrategy.noWatermarks(), "doris source").print()

    env.execute()
  }

  /**
   * json 写 doris
   */
  def writeAppend(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val optionBuilder = DorisOptions.builder()
    optionBuilder.setFenodes("192.168.206.130:8030")
      .setTableIdentifier("test_db.flinktest")
      .setUsername("root")
      .setPassword("")

    val executionBuilder = DorisExecutionOptions.builder()
    val properties = new Properties()
    properties.setProperty("format", "json")
    properties.setProperty("read_json_by_line", "true")
    executionBuilder.setLabelPrefix("label-doris")
      .setDeletable(false)
      .setStreamLoadProp(properties)

    val sinkBuilder: DorisSink.Builder[String] = DorisSink.builder()
    sinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
      .setDorisExecutionOptions(executionBuilder.build())
      .setSerializer(new SimpleStringSerializer())
      .setDorisOptions(optionBuilder.build())

    env.fromElements("{\"siteid\":\"10\", \"citycode\": \"1001\",\"username\": \"ww\",\"pv\":\"100\"}")
      .sinkTo(sinkBuilder.build())

    env.execute()
  }

  /**
   * RowData 写 Doris
   */
  def writeAppend2(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val optionBuilder = DorisOptions.builder()
    optionBuilder.setFenodes("192.168.206.130:8030")
      .setTableIdentifier("test_db.flinktest")
      .setUsername("root")
      .setPassword("")

    val executionBuilder = DorisExecutionOptions.builder()
    val properties = new Properties()
    properties.setProperty("format", "json")
    properties.setProperty("read_json_by_line", "true")
    executionBuilder.setLabelPrefix("label-doris-w")
      .setDeletable(false)
      .setStreamLoadProp(properties)

    val fields = Array("siteid", "citycode", "username", "pv")
    val types = Array(DataTypes.INT(), DataTypes.SMALLINT(), DataTypes.VARCHAR(255), DataTypes.BIGINT())
    val sinkBuilder: DorisSink.Builder[RowData] = DorisSink.builder()
    sinkBuilder.setDorisReadOptions(DorisReadOptions.builder().build())
      .setDorisExecutionOptions(executionBuilder.build())
      .setSerializer(RowDataSerializer
        .builder()
        .setFieldNames(fields)
        .setFieldType(types)
        .setType("json")
        .build())
      .setDorisOptions(optionBuilder.build())


    env.fromElements("{\"siteid\":\"1010\", \"citycode\": \"1002\",\"username\": \"wang\",\"pv\":\"100\"}")
      .map(new MapFunction[String, RowData] {
        override def map(json: String): GenericRowData = {
          val obj = JSON.parseObject(json)
          val rowData: GenericRowData = new GenericRowData(4);
          rowData.setField(0, obj.getIntValue("siteid"))
          rowData.setField(1, obj.getShortValue("citycode"))
          rowData.setField(2, StringData.fromString(obj.getString("username")))
          rowData.setField(3, obj.getLongValue("pv"))
          rowData
        }
      }).sinkTo(sinkBuilder.build())

    env.execute()
  }

  /**
   * sql 写 Doris
   */
  def writeAppend3(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val tEnv = StreamTableEnvironment.create(env)
    tEnv.executeSql("create table flink_0518(" +
      " siteid int, " +
      " citycode int, " +
      " username string, " +
      " pv bigint " +
      ")with(" +
      "  'connector' = 'doris', " +
      "  'fenodes' = '192.168.206.130:8030', " +
      "  'table.identifier' = 'test_db.flinktest', " +
      "  'username' = 'root', " +
      "  'password' = '' " +
      ")")

    tEnv.executeSql("insert into flink_0518(siteid, citycode, username, pv) values(4, 4, 'wangting', 4) ")
  }

  /**
   * sql 读 Doris
   */
  def readAll2(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val tEnv = StreamTableEnvironment.create(env)
    tEnv.executeSql("create table flink_0518(" +
      " siteid int, " +
      " citycode smallint, " +
      " username string, " +
      " pv bigint " +
      ")with(" +
      "  'connector' = 'doris', " +
      "  'fenodes' = '192.168.206.130:8030', " +
      "  'table.identifier' = 'test_db.flinktest', " +
      "  'username' = 'root', " +
      "  'password' = '' " +
      ")")

    tEnv.sqlQuery("select * from flink_0518").execute().print()
  }

  def main(args: Array[String]): Unit = {


  }

}
